﻿namespace WPFDBBrowser.MessageBus
{
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using System.Threading;
    using Amib.Threading;

    public class MessageBus : IMessageBus
    {

        private Dictionary<string, HavingStatus<IEndPoint, EndpointStatus>> endpoints;
        private List<ISubscription> subscriptions;
        private SmartThreadPool threadPool;
        //private ILog log;
        private object endpointsLock = new object();
        private object subscriptionsLock = new object();
        //private object threadPoolLock = new object();
        //private object logLock = new object();

        private Dictionary<string, IWorkItemsGroup> workItemGroups;

        public MessageBus()
        {
            endpoints = new Dictionary<string, HavingStatus<IEndPoint, EndpointStatus>>();
            subscriptions = new List<ISubscription>();
            threadPool = new SmartThreadPool(1000, 4);
            //log = LogManager.GetLogger("Porthus.THC.MessageBus");
            workItemGroups = new Dictionary<string, IWorkItemsGroup>();
        }

        #region IMessageBus Members

        public IEndPoint RegisterEndpoint(string address)
        {
            IEndPoint result = null;
            lock (endpointsLock)
            {
                if (!endpoints.ContainsKey(address))
                {
                    endpoints.Add(
                        address,
                        new HavingStatus<IEndPoint, EndpointStatus>
                        {
                            Value = new EndPoint { Address = address, Bus = this },
                            Status = EndpointStatus.Stopped
                        });
                    workItemGroups.Add(address, threadPool.CreateWorkItemsGroup(2));
                    //log.Debug(String.Format("Endpoint '{0}' registered - {1}", address, Thread.CurrentThread.GetHashCode()));
                }
                result = endpoints[address].Value;
            }
            return result;
        }

        public void UnRegisterEndpoint(string address)
        {
            lock (endpointsLock)
            {
                if (endpoints.ContainsKey(address))
                {
                    endpoints.Remove(address);
                    workItemGroups.Remove(address);
                    //log.Debug(String.Format("Endpoint '{0}' unregistered - {1}", address, Thread.CurrentThread.GetHashCode()));

                    lock (subscriptionsLock)
                    {
                        List<ISubscription> subscriptionsToRemove = new List<ISubscription>(from s in subscriptions where s.Subscriber == address select s);

                        foreach (var item in subscriptionsToRemove)
                        {
                            subscriptions.Remove(item);
                            //log.Debug(String.Format("Endpoint '{0}' unsubscribed subscription [{1}] - {2}", item.Subscriber, item.Id, Thread.CurrentThread.GetHashCode()));
                        }
                    }

                }
            }
        }

        public IEndPoint Start(string address)
        {
            IEndPoint result = null;
            lock (endpointsLock)
            {
                if (endpoints.ContainsKey(address))
                {
                    endpoints[address].Status = EndpointStatus.Started;
                    //log.Debug(String.Format("Endpoint '{0}' started - {1}", address, Thread.CurrentThread.GetHashCode()));
                    result = endpoints[address].Value;
                }
            }
            return result;
        }

        public IEndPoint Stop(string address)
        {
            IEndPoint result = null;
            lock (endpointsLock)
            {
                if (endpoints.ContainsKey(address))
                {
                    endpoints[address].Status = EndpointStatus.Stopped;
                    //log.Debug(String.Format("Endpoint '{0}' stopped - {1}", address, Thread.CurrentThread.GetHashCode()));
                    return endpoints[address].Value;
                }
            }
            return result;
        }

        public IEndPoint Publish<TMessage>(string publisher, TMessage message)
        {
            IEndPoint result = null;
            lock (endpointsLock)
            {
                if (endpoints.ContainsKey(publisher) && endpoints[publisher].Status == EndpointStatus.Started)
                {
                    //log.Debug(String.Format("Endpoint '{0}' published message '{1}' - {2}", publisher, message.GetType().Name, Thread.CurrentThread.GetHashCode()));
                    lock (subscriptionsLock)
                    {
                        var candidates =
                            (from item in subscriptions.OfType<Subscription<TMessage>>()
                             where endpoints.ContainsKey(item.Subscriber)
                             && endpoints[item.Subscriber].Status == EndpointStatus.Started
                             && ((item.Filter == null) || (item.Filter(message)))
                             select item).ToList();

                        // Add metadata to context
                        //if (message is IContext<Dictionary<string, string>>)
                        //{
                        //    ((IContext<Dictionary<string, string>>)message).Context[MessageContextProperties.Source] = publisher;
                        //}

                        foreach (var candidate in candidates)
                        {
                            //log.Debug(String.Format("Queued message '{0}' for endpoint '{1}' [{2}/{3}] - {4}", message.GetType().Name, candidate.Subscriber, threadPool.InUseThreads, threadPool.MaxThreads, Thread.CurrentThread.GetHashCode()));
                            workItemGroups[candidate.Subscriber].QueueWorkItem(delegate(TMessage msg) { candidate.Notify(msg); }, message);
                        }
                        result = endpoints[publisher].Value;
                    }
                }
            }
            return result;
        }

        public IEndPoint Subscribe<TMessage>(string subscriber, Guid subscriptionId, Action<TMessage> notify)
        {
            return Subscribe<TMessage>(subscriber, subscriptionId, notify, null);
        }

        public IEndPoint Subscribe<TMessage>(string subscriber, Guid subscriptionId, Action<TMessage> notify, Predicate<TMessage> filter)
        {
            IEndPoint result = null;
            lock (endpointsLock)
            {
                if (endpoints.ContainsKey(subscriber))
                {
                    lock (subscriptionsLock)
                    {
                        subscriptions.Add(new Subscription<TMessage>
                        {
                            Id = subscriptionId,
                            Subscriber = subscriber,
                            Notify = notify,
                            Filter = filter
                        });

                    }
                    //log.Debug(String.Format("Endpoint '{0}' subscribed to message '{1}' [{2}] - {3}", subscriber, typeof(TMessage).Name, subscriptionId, Thread.CurrentThread.GetHashCode()));
                    result = endpoints[subscriber].Value;
                }
            }
            return result;
        }

        public IEndPoint Send<TMessage>(string sender, TMessage message, string recipient)
        {
            throw new NotImplementedException();
        }

        public IEndPoint Send<TRequest, TResponse>(string sender, TRequest request, string recipient, Action<TResponse> reply)
        {
            throw new NotImplementedException();
        }

        public IEndPoint UnSubscribe(Guid subscriptionId)
        {
            IEndPoint result = null;
            lock (endpointsLock)
            {
                lock (subscriptionsLock)
                {
                    var searchResult = subscriptions.Where(s => s.Id == subscriptionId).FirstOrDefault();
                    if (searchResult != null)
                    {
                        subscriptions.Remove(searchResult);
                        //log.Debug(String.Format("Endpoint '{0}' unsubscribed subscription [{1}] - {2}", searchResult.Subscriber, subscriptionId, Thread.CurrentThread.GetHashCode()));
                        result = endpoints[searchResult.Subscriber].Value;
                    }
                }
            }
            return result;
        }

        #endregion
    }
}
